use crate::chunk::Chunk;
use crate::index::IndexStore;
use crate::recipe::FileInfo;
use crate::container::{ContainerStore,RecentChunk};
use super::{BackupStatistics, BackupResult, BackupConfig};
use super::fingerprint::HasherProduct;
use crate::constants::BACKUP_DEDUP_LOG;
use crate::recipe::pointer::ChunkPointer;

use std::sync::Arc;
use std::time::Instant;
use parking_lot::Mutex;
use pipelines::{Sender,Receiver};

pub struct ChunkDeduper {
    container_store:Arc<ContainerStore>,
    index_store:Arc<IndexStore>,
    statistics:Arc<Mutex<BackupStatistics>>,
    find_in_recent_buffer:bool,
    find_in_read_cache:bool,
    find_in_write_buf:bool,
    find_in_index_store:bool,
}


impl ChunkDeduper {
    pub fn new(container_store:Arc<ContainerStore>,index_store:Arc<IndexStore>,statistics:Arc<Mutex<BackupStatistics>>,config:&BackupConfig) -> Self {
        Self {
            container_store:container_store,
            index_store:index_store,
            statistics:statistics,
            find_in_recent_buffer:config.find_in_recent_buffer,
            find_in_read_cache:config.find_in_read_cache,
            find_in_write_buf:config.find_in_write_buf,
            find_in_index_store:config.find_in_index_store,
        }
    }

	pub fn dedup<P:ChunkPointer+Send>(&self,out:&Sender<DeduperProduct<P>>,recv:Receiver<HasherProduct>) -> BackupResult<()> {
        //let mut chunk_counter:usize = 0;
        let mut recent_buffer_time:Option<u128> = self.find_in_recent_buffer.then_some(0);
        let mut read_cache_time:Option<u128> = self.find_in_read_cache.then_some(0);
        let mut write_buffer_time:Option<u128> = self.find_in_write_buf.then_some(0);
        let mut index_store_time:Option<u128> = self.find_in_index_store.then_some(0);
        let mut insert_to_recent_buffer_time:Option<u128> = self.find_in_recent_buffer.then_some(0);
        for product in recv {
            match product {
                HasherProduct::FileInfo(file_info) => {
                    out.send(DeduperProduct::FileInfo(file_info));
                }
                HasherProduct::Chunk(mut chunk) => {
                    let (dedup_in_recent_buffer_time,find) = self.dedup_in_recent_buffer(&mut chunk, out)?;
                    recent_buffer_time = recent_buffer_time.map(|val|val + dedup_in_recent_buffer_time);
                    if find { continue; }

                    let (dedup_in_read_cache_time,find) = self.dedup_in_read_cache(&chunk, out)?;
                    read_cache_time = read_cache_time.map(|val|val + dedup_in_read_cache_time);
                    if find { continue; }

                    let (dedup_in_write_buf_time,find) = self.dedup_in_write_buf(&chunk, out)?;
                    write_buffer_time = write_buffer_time.map(|val|val + dedup_in_write_buf_time);
                    if find { continue; }

                    let (dedup_in_index_store_time,find) = self.dedup_in_index_store(&chunk, out)?;
                    index_store_time = index_store_time.map(|val|val + dedup_in_index_store_time);
                    if find { continue; }
                
                    let now2 = Instant::now();
                    self.container_store.insert_into_recent_buffer(chunk.fp(), RecentChunk::Chunk(chunk.clone()));
                    insert_to_recent_buffer_time = insert_to_recent_buffer_time.map(|val| val + now2.elapsed().as_micros());
                    out.send(DeduperProduct::Chunk(chunk));
                    //chunk_counter += 1;
                }
            }
        }
        let cost_time = 
            recent_buffer_time.map_or(0,|v|v) +
            read_cache_time.map_or(0,|v|v) +
            write_buffer_time.map_or(0,|v|v) +
            index_store_time.map_or(0,|v|v) +
            insert_to_recent_buffer_time.map_or(0,|v|v);
        self.statistics.lock().set_dedup_phase_time(cost_time);
        Ok(())
    }

    fn dedup_in_recent_buffer<P:ChunkPointer+Send>(&self,chunk:&mut Arc<Chunk>,out:&Sender<DeduperProduct<P>>) -> BackupResult<(u128,bool)> {
        if !self.find_in_recent_buffer { return Ok((0,false)); }
        let mut find = false;
        let mut cost_time:u128 = 0;
        let now = Instant::now();
        if let Some(recent) = self.container_store.find_in_recent_buffer(chunk.fp()) {
            log::debug!(target:BACKUP_DEDUP_LOG,"chunk {:?} find duplicate chunk in recent buffer\n",chunk.fp());
            match recent {
                RecentChunk::Chunk(dup_chunk) => {
                    Arc::get_mut(chunk).expect("REASON").set_duplicate(&dup_chunk);
                    //container_store.insert_into_recent_buffer(fp, RecentChunk::Chunk(chunk.clone()));
                    find = true;
                    cost_time += now.elapsed().as_micros();
                    out.send(DeduperProduct::Chunk(chunk.clone()));
                }
                RecentChunk::ChunkPointer(id,no) => {
                    let cp = Box::new(P::new(id,no,chunk.fp()));
                    find = true;
                    cost_time += now.elapsed().as_micros();
                    out.send(DeduperProduct::ChunkPointer(cp));
                }
            }
            //chunk_counter += 1;
        }
        Ok((cost_time,find))
    }

    fn dedup_in_read_cache<P:ChunkPointer+Send>(&self,chunk:&Arc<Chunk>,out:&Sender<DeduperProduct<P>>) -> BackupResult<(u128,bool)> { 
        if !self.find_in_read_cache { return Ok((0,false)); }
        let mut find = false;
        let mut cost_time:u128 = 0;
        let now = Instant::now();
        if let Some(cp) = self.container_store.find_chunk_pointer_in_read_cache::<P>(chunk.fp()) {
            log::debug!(target:BACKUP_DEDUP_LOG,"chunk {:?} find duplicate chunk in read cache\n",chunk.fp());
            //container_store.insert_into_recent_buffer(fp, RecentChunk::ChunkPointer(cp.containerid(), cp.no()));
            find = true;
            cost_time += now.elapsed().as_micros();
            out.send(DeduperProduct::ChunkPointer(cp));
            //chunk_counter += 1;
        }
        Ok((cost_time,find))
    }

    fn dedup_in_write_buf<P:ChunkPointer+Send>(&self,chunk:&Arc<Chunk>,out:&Sender<DeduperProduct<P>>) -> BackupResult<(u128,bool)> { 
        if !self.find_in_write_buf { return Ok((0,false)); }
        let mut find = false;
        let mut cost_time:u128 = 0;
        let now = Instant::now();
        if let Some(cp) = self.container_store.find_chunk_pointer_in_write_buf::<P>(chunk.fp()) {
            log::debug!(target:BACKUP_DEDUP_LOG,"chunk {:?} find duplicate chunk in write buffer\n",chunk.fp());
            //container_store.insert_into_recent_buffer(fp, RecentChunk::ChunkPointer(cp.containerid(), cp.no()));
            find = true;
            cost_time += now.elapsed().as_micros();
            out.send(DeduperProduct::ChunkPointer(cp));
            //chunk_counter += 1;
        }
        return Ok((cost_time,find));
    }

    fn dedup_in_index_store<P:ChunkPointer+Send>(&self,chunk:&Arc<Chunk>,out:&Sender<DeduperProduct<P>>) -> BackupResult<(u128,bool)> { 
        if !self.find_in_index_store { return Ok((0,false)); }
        let mut find = false;
        let mut cost_time:u128 = 0;
        let now = Instant::now();
        if let Some(cp) = self.index_store.find_with_fp::<P>(chunk.fp())? {
            log::debug!(target:BACKUP_DEDUP_LOG,"chunk {:?} find duplicate chunk in kvstore\n",chunk.fp());
            //container_store.insert_into_recent_buffer(fp, RecentChunk::ChunkPointer(cp.containerid(), cp.no()));
            find = true;
            cost_time += now.elapsed().as_micros();
            out.send(DeduperProduct::ChunkPointer(cp));
            //chunk_counter += 1;
        }
        return Ok((cost_time,find));
    }
}

pub enum DeduperProduct<P:ChunkPointer+Send> {
	FileInfo(FileInfo),
	ChunkPointer(Box<P>),
	Chunk(Arc<Chunk>),
}